package com.gjy.kafka.sp.service;

import com.gjy.kafka.sp.config.KafkaConstant;
import com.gjy.kafka.sp.entity.MessageEntity;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.Objects;

/**
 * Kafka消息消费者服务
 * 负责从Kafka接收并处理消息
 *
 * @author gjy
 * @version 1.0
 * @since 2025-09-07 10:46:57
 */
@Service
public class KafkaConsumerService {
    private static final Logger log = LoggerFactory.getLogger(KafkaConsumerService.class);
    /**
     * 消息轨迹服务
     */
    private final MessageTraceService messageTraceService;

    public KafkaConsumerService(MessageTraceService messageTraceService) {
        this.messageTraceService = messageTraceService;
    }

    /**
     * 消费普通消息
     *
     * @param record         消息记录
     * @param acknowledgment 确认对象
     * @param topic          主题名称
     * @param partition      分区号
     * @param offset         偏移量
     */
    @KafkaListener(topics = KafkaConstant.NORMAL_TOPIC, groupId = KafkaConstant.NORMAL_CONSUMER_GROUP)
    public void consumeNormalMessage(ConsumerRecord<String, MessageEntity> record,
                                     Acknowledgment acknowledgment,
                                     @Header("kafka_receivedTopic") String topic,
                                     @Header("kafka_receivedPartitionId") int partition,
                                     @Header("kafka_offset") long offset) {
        MessageEntity message = record.value();
        Objects.requireNonNull(message, "消息内容不能为空");

        log.info("接收到普通消息，主题：{}，分区：{}，偏移量：{}，消息ID：{}，业务类型：{}",
                topic, partition, offset, message.getMessageId(), message.getBusinessType());

        try {
            // 处理消息的业务逻辑
            processMessage(message);

            // 记录消费成功轨迹
            messageTraceService.recordConsumeSuccess(message.getMessageId(), partition, offset);

            // 手动确认消息
            acknowledgment.acknowledge();
            log.info("普通消息处理成功并确认，主题：{}，消息ID：{}", topic, message.getMessageId());
        } catch (Exception e) {
            // 记录消费失败轨迹
            messageTraceService.recordConsumeFailure(message.getMessageId(), partition, offset, e.getMessage());

            log.error("普通消息处理失败，主题：{}，消息ID：{}", topic, message.getMessageId(), e);

            // 手动确认消息（将失败消息标记为已消费，避免无限重试）
            // 如果需要将消息发送到死信队列，可以不确认并配置死信转发
            acknowledgment.acknowledge();
        }
    }

    /**
     * 消费分区消息
     *
     * @param record         消息记录
     * @param acknowledgment 确认对象
     * @param topic          主题名称
     * @param partition      分区号
     * @param offset         偏移量
     */
    @KafkaListener(topics = KafkaConstant.PARTITION_TOPIC, groupId = KafkaConstant.PARTITION_CONSUMER_GROUP)
    public void consumePartitionMessage(ConsumerRecord<String, MessageEntity> record,
                                        Acknowledgment acknowledgment,
                                        @Header("kafka_receivedTopic") String topic,
                                        @Header("kafka_receivedPartitionId") int partition,
                                        @Header("kafka_offset") long offset) {
        MessageEntity message = record.value();
        Objects.requireNonNull(message, "消息内容不能为空");

        log.info("接收到分区消息，主题：{}，分区：{}，偏移量：{}，消息ID：{}，业务ID：{}，业务类型：{}",
                topic, partition, offset, message.getMessageId(), message.getBusinessId(), message.getBusinessType());

        try {
            // 处理消息的业务逻辑
            processMessage(message);

            // 记录消费成功轨迹
            messageTraceService.recordConsumeSuccess(message.getMessageId(), partition, offset);

            // 手动确认消息
            acknowledgment.acknowledge();
            log.info("分区消息处理成功并确认，主题：{}，消息ID：{}", topic, message.getMessageId());
        } catch (Exception e) {
            // 记录消费失败轨迹
            messageTraceService.recordConsumeFailure(message.getMessageId(), partition, offset, e.getMessage());

            log.error("分区消息处理失败，主题：{}，消息ID：{}", topic, message.getMessageId(), e);
            acknowledgment.acknowledge();
        }
    }

    /**
     * 消费事务消息
     *
     * @param record         消息记录
     * @param acknowledgment 确认对象
     * @param topic          主题名称
     * @param partition      分区号
     * @param offset         偏移量
     */
    @Transactional(rollbackFor = Exception.class)
    @KafkaListener(topics = KafkaConstant.TRANSACTIONAL_TOPIC, groupId = KafkaConstant.TRANSACTIONAL_CONSUMER_GROUP)
    public void consumeTransactionalMessage(ConsumerRecord<String, MessageEntity> record,
                                            Acknowledgment acknowledgment,
                                            @Header("kafka_receivedTopic") String topic,
                                            @Header("kafka_receivedPartitionId") int partition,
                                            @Header("kafka_offset") long offset) {
        MessageEntity message = record.value();
        Objects.requireNonNull(message, "消息内容不能为空");

        log.info("接收到事务消息，主题：{}，分区：{}，偏移量：{}，消息ID：{}，业务类型：{}",
                topic, partition, offset, message.getMessageId(), message.getBusinessType());

        try {
            // 处理消息的业务逻辑
            processMessage(message);

            // 这里可以添加数据库操作等其他事务操作

            // 记录消费成功轨迹
            messageTraceService.recordConsumeSuccess(message.getMessageId(), partition, offset);

            // 手动确认消息
            acknowledgment.acknowledge();
            log.info("事务消息处理成功并确认，主题：{}，消息ID：{}", topic, message.getMessageId());
        } catch (Exception e) {
            // 记录消费失败轨迹
            messageTraceService.recordConsumeFailure(message.getMessageId(), partition, offset, e.getMessage());

            log.error("事务消息处理失败，主题：{}，消息ID：{}", topic, message.getMessageId(), e);
            // 事务会回滚，消息不会被确认，将被重新消费
        }
    }

    /**
     * 消费死信消息
     *
     * @param record         消息记录
     * @param acknowledgment 确认对象
     * @param topic          主题名称
     * @param partition      分区号
     * @param offset         偏移量
     */
    @KafkaListener(topics = KafkaConstant.DEAD_LETTER_TOPIC, groupId = KafkaConstant.DEAD_LETTER_CONSUMER_GROUP)
    public void consumeDeadLetterMessage(ConsumerRecord<String, MessageEntity> record,
                                         Acknowledgment acknowledgment,
                                         @Header("kafka_receivedTopic") String topic,
                                         @Header("kafka_receivedPartitionId") int partition,
                                         @Header("kafka_offset") long offset) {
        MessageEntity message = record.value();
        Objects.requireNonNull(message, "消息内容不能为空");

        log.error("接收到死信消息，主题：{}，分区：{}，偏移量：{}，消息ID：{}，业务类型：{}",
                topic, partition, offset, message.getMessageId(), message.getBusinessType());

        try {
            // 处理死信消息的业务逻辑，通常需要人工干预
            processDeadLetterMessage(message);

            // 记录消费成功轨迹
            messageTraceService.recordConsumeSuccess(message.getMessageId(), partition, offset);

            // 手动确认消息
            acknowledgment.acknowledge();
            log.info("死信消息处理成功并确认，主题：{}，消息ID：{}", topic, message.getMessageId());
        } catch (Exception e) {
            // 记录消费失败轨迹
            messageTraceService.recordConsumeFailure(message.getMessageId(), partition, offset, e.getMessage());

            log.error("死信消息处理失败，主题：{}，消息ID：{}", topic, message.getMessageId(), e);
            acknowledgment.acknowledge();
        }
    }

    /**
     * 处理消息的业务逻辑
     *
     * @param message 要处理的消息
     */
    private void processMessage(MessageEntity message) {
        // 根据业务类型处理不同的消息
        String businessType = message.getBusinessType();
        if ("ORDER_CREATE".equals(businessType)) {
            // 处理订单创建消息
            processOrderCreateMessage(message);
        } else if ("USER_REGISTER".equals(businessType)) {
            // 处理用户注册消息
            processUserRegisterMessage(message);
        } else {
            // 处理未知类型消息
            log.warn("收到未知类型的消息，消息ID：{}，业务类型：{}",
                    message.getMessageId(), businessType);
        }
    }

    /**
     * 处理死信消息
     *
     * @param message 死信消息
     */
    private void processDeadLetterMessage(MessageEntity message) {
        log.info("处理死信消息，消息ID：{}，内容：{}",
                message.getMessageId(), message.getContent());
        // 实际业务处理逻辑，如记录到数据库等待人工处理
    }

    /**
     * 处理订单创建消息
     *
     * @param message 订单创建消息
     */
    private void processOrderCreateMessage(MessageEntity message) {
        log.info("处理订单创建消息，消息ID：{}，订单信息：{}",
                message.getMessageId(), message.getContent());
        // 实际业务处理逻辑...
    }

    /**
     * 处理用户注册消息
     *
     * @param message 用户注册消息
     */
    private void processUserRegisterMessage(MessageEntity message) {
        log.info("处理用户注册消息，消息ID：{}，用户信息：{}",
                message.getMessageId(), message.getContent());
        // 实际业务处理逻辑...
    }

}
